package jcu.salt;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import jcu.sal.common.exceptions.ConfigurationException;
import jcu.salt.client.DataHandler;
import jcu.salt.client.SaltRmiClient;
import jcu.salt.client.SaltSRBClient;
import jcu.salt.io.Receiver;
import jcu.salt.io.Sender;
import jcu.salt.io.Transmittable;

import com.rbnb.sapi.ChannelMap;
import com.rbnb.sapi.SAPIException;
import com.rbnb.sapi.Sink;

/**
 * This class is used to <br>
 * - form the glue between SAL units and the data turbine, and <br>
 * - form the glue between data turbine and SRB.
 * 
 * @author SAL-T Team 2008
 * 
 */
public class SALT implements DataHandler {

	/**
	 * This is the startup of SAL-T
	 * 
	 * @param args
	 */
	public static void main(String[] args) {

		SALT s = new SALT();
		if (args.length == 2 && args[0].equals("send")) {
			s.send(args[1]);
		} else if (args.length == 2 && args[0].equals("recv")) {
			s.recv(args[1]);
		} else if (args.length == 1 && args[0].equals("sal")) {
			s.sal();
		} else if (args.length == 1 && args[0].equals("srb")) {
			s.srb();
		} else if (args.length == 1 && args[0].equals("channelview")) {
			s.channelview();
		} else if (args.length == 1 && args[0].equals("multisink")) {
			s.multisink();
		} else if (args.length == 5 && args[0].equals("stress")) {
			s.stressTest(args[1], Integer.parseInt(args[2]), Integer
					.parseInt(args[3]), args[4]);
		} else {
			System.err
					.println("usage: SALT sal | send <str> | recv <channel_name> |"
							+ " stress <connprefix> <#connections> <#packages> <str> |"
							+ "channelview | multisink");
		}
	}

	private void multisink() {
		Logger.getInstance().setOutput(System.out); // enable logging
		Configuration config = Configuration.getInstance();
		
		Map<Sink,ChannelMap> sinkmap = new HashMap<Sink, ChannelMap>();
		
		ChannelMap fullcmap = getSALTChannels();

		Logger.log("init sinks ");
		int i = 0;
		for (String str : fullcmap.GetChannelList()) {
			Logger.log(++i + " : init for source channel " + str);
			
			Sink sink = new Sink();
			ChannelMap cmap = new ChannelMap();
			try {
				sink.OpenRBNBConnection(config.get(Configuration.DATATURBINE_IP),
						config.get(Configuration.SINK));				
				cmap.Add(str);
				
				sink.Subscribe(cmap);
				
				sinkmap.put(sink, cmap);
			} catch (SAPIException e) {
				System.err.println("failed to init for "+ str);
				e.printStackTrace();
			}			
		}
		
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		for (Sink sink : sinkmap.keySet()) {
			Logger.log("close rbnb connection");
			sink.CloseRBNBConnection();			
		}
		
		
	}

	private ChannelMap getSALTChannels() {
		// fetch configuration
		Configuration config = Configuration.getInstance();

		ChannelMap cmap = new ChannelMap();
		Sink sink = new Sink();

		try {
			cmap.Add("...");

			sink.OpenRBNBConnection(config.get(Configuration.DATATURBINE_IP),
					config.get(Configuration.SINK) + "_CHANNELVIEW");
			sink.RequestRegistration(cmap); // get all registered channels
			// matching the cmap
			sink.Fetch(2000, cmap); // get result of the request registration
			// method and overwrite cmap

			sink.CloseRBNBConnection(); // close connection

			ChannelMap filtered = new ChannelMap();
			String srcprefix = config.get(Configuration.SOURCE);
			String channelprefix = config.get(Configuration.DEFAULTCHANNEL);

			for (String str : cmap.GetChannelList()) {
				if (str.matches("^" + srcprefix + ".*/" + channelprefix + ".*")) {
					filtered.Add(str);
				}
			}
			cmap = filtered;

		} catch (SAPIException e) {
			e.printStackTrace();
		}

		return cmap;
	}

	/**
	 * print all open channels of the configured source to stdout
	 */
	private void channelview() {

		ChannelMap cmap = getSALTChannels();

		System.out.println("Open Channels : ");
		int i = 0;
		for (String str : cmap.GetChannelList()) {
			System.out.println(++i + " : " + str);
		}
	}

	/**
	 * This method is directly called by a specific command line argument to
	 * receive data from data turbine and pass it on into SRB.
	 * 
	 */
	private void srb() {

		// fetch configuration
		Configuration config = Configuration.getInstance();

		Logger.getInstance().setOutput(System.out); // enable logging

		SaltSRBClient srb = new SaltSRBClient(config
				.get(Configuration.DATATURBINE_IP), config
				.get(Configuration.SINK), getSALTChannels());
		srb.start();
		menu();
		srb.stop();
	}

	/**
	 * starts up the glue between SAL and DataTurbine
	 */
	public void sal() {

		Logger.getInstance().setOutput(System.out); // enable logging

		Configuration config = Configuration.getInstance();

		Logger.log("make use of client IP "
				+ config.get(Configuration.CLIENT_IP));
		Logger
				.log("make use of agent IP "
						+ config.get(Configuration.AGENT_IP));
		try {
			final SaltRmiClient c = new SaltRmiClient(config
					.get(Configuration.CLIENT_NAME), config
					.get(Configuration.AGENT_IP), config
					.get(Configuration.CLIENT_IP));
			c.start(config.get(Configuration.CLIENT_IP));
			c.run();

			menu(); // waiting for user input

			c.stop();
		} catch (RemoteException e) {
			System.err.println("Failed to stop properly");
		} catch (ConfigurationException e) {
			e.printStackTrace();
		}
		Logger.log("SALT finished");
		System.exit(0);
	}

	/**
	 * provides a menu which enables the user to interact with the system during
	 * runtime
	 * 
	 * TODO improve menu
	 */
	private void menu() {
		InputStreamReader in = new InputStreamReader(System.in);
		BufferedReader reader = new BufferedReader(in);
		try {
			reader.readLine();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void recv(String channel) {
		Logger.getInstance().setOutput(System.out); // enable logging
		Configuration config = Configuration.getInstance();

		final Receiver rec = new Receiver(config.get(Configuration.DATATURBINE_IP),
				config.get(Configuration.SOURCE), config
						.get(Configuration.SINK),
						channel);

		Logger.log(" +++ connect +++");
		rec.connect();
		
		Logger.log(" +++ listen +++ (using own thread)");
		
		RecThread t = new RecThread(this, rec);
		t.start();		
		Logger.log(" +++ started +++ ");
		
		menu(); // wait for usrin
		
		System.out.println(" +++ stop and disconnect +++");
		t.disconnect();
		try {
			t.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	private class RecThread extends Thread {

		private DataHandler hdlr;
		private Receiver rec;
		public RecThread(DataHandler hdlr, Receiver rec) {
			this.hdlr = hdlr;
			this.rec = rec;
		}
		
		@Override
		public void run() {
			rec.receive(hdlr);
			super.run();
		}
		
		public void disconnect() {
			rec.setActive(false);
			rec.disconnect();
		}
		
	}

	/**
	 * Send data to data turbine, which was set in the constructor of this
	 * class.
	 * 
	 * @param data
	 *            the data to send
	 */
	public void send(String data) {
		Configuration config = Configuration.getInstance();

		Sender sender = new Sender(config.get(Configuration.DATATURBINE_IP),
				config.get(Configuration.SOURCE), config
						.get(Configuration.DEFAULTCHANNEL));

		sender.connect();
		sender.send(data);
		sender.disconnect();
	}

	/**
	 * Method to stress test the data turbine connection.
	 * 
	 * @param connectionPrefix
	 *            Connections are named <prefix> + number
	 * @param numberOfConnections
	 *            How many connections should be opened to data turbine.
	 * @param numberOfPackages
	 *            How often the data is sent.
	 * @param data
	 *            the data to send
	 */
	public void stressTest(String connectionPrefix, int numberOfConnections,
			int numberOfPackages, String data) {

		Logger.getInstance().setOutput(System.out); // enable logging

		Configuration config = Configuration.getInstance();
		ArrayList<Sender> senders = new ArrayList<Sender>();

		// create data turbine senders
		Logger.log("try to establish " + numberOfConnections + " connectons to turbine");
		for (int i = 0; i < numberOfConnections; i++) {
			senders.add(new Sender(config.get(Configuration.DATATURBINE_IP),
					config.get(Configuration.SOURCE), connectionPrefix + i));
		}

		// connect
		for (Sender s : senders) {
			s.connect();
		}

		// send specified amount of packages
		for (int i = 0; i < numberOfPackages; i++) {
			// .. to each connection
			Logger.log("try to send package #" + (int)(i+1) +" of " + numberOfPackages + " to all " + senders.size() + " channels");
			for (Sender s : senders) {
				s.send(data);
			}			
		}

		// connect
		for (Sender s : senders) {
			s.disconnect();
		}
	}

	@Override
	public void handleData(Transmittable received) {		
		System.out.println("Received : " + received);
		System.out.println("msg = [" + new String(received.getData()) + "]");

	}
}
